package io.mqttpush.mqttserver.msgs;


import io.mqttpush.mqttserver.config.DeviceManager;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collection;
import java.util.Map;

/**
 * @author tianzhenjiu
 * @date 2020/4/27 16:46
 */
public class MessageSender {


    protected Logger logger = LoggerFactory.getLogger(getClass());

    static MessageSender messageSender=new MessageSender();



    public  static MessageSender getInstance() {

        return messageSender;
    }




    public void sendMessage(String from,String to,MqttPublishMessage message){

        logger.info("{} send msg 2{}",from,to);
        DeviceManager deviceManager=DeviceManager.getDeviceManager();

        Collection<Channel> channels= MessageRoute.getInstance().routeTopic(to);

        if(channels==null||channels.isEmpty()){
            logger.info("send to is empty{}",to);
            return;
        }


        channels.forEach((channel)->{

            message.retain();

            message.fixedHeader().qosLevel();


            Map<String,MqttQoS> mqttQoSMap=channel.attr(deviceManager.channelTopics).get();
            MqttQoS mqttQoS=null;
            if(mqttQoSMap.containsKey(to)){
                mqttQoS=mqttQoSMap.get(to);
            }


            if(mqttQoS==null){
                mqttQoS=MqttQoS.AT_LEAST_ONCE;
            }

            ByteBuf byteBuf= ByteBufAllocator.DEFAULT.buffer();

            byteBuf.writeBytes(mqttQoS.toString().getBytes());

            channel.writeAndFlush( createCommonMqttPublish(mqttQoS,to,message.content()));

        });

    }


    /**
     * 新建应该一般性的mqtt消息
     * @param subQos
     * @param topicName
     * @param payload
     * @return
     */
    public  MqttPublishMessage createCommonMqttPublish(MqttQoS subQos,String topicName,ByteBuf payload){



        int packId=1;

        MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(
                MqttMessageType.PUBLISH,
                false,
                subQos,false, 0);

        MqttPublishVariableHeader variableHeader = new MqttPublishVariableHeader(topicName,
                packId);



        return  new MqttPublishMessage(mqttFixedHeader, variableHeader, payload);
    }

    public  MqttPublishMessage copyAPublishMsg(MqttQoS subQos,MqttPublishMessage message){


        Boolean isDump=message.fixedHeader().isDup();
        Boolean isRetain=message.fixedHeader().isDup();
        String topicName=message.variableHeader().topicName();
        int packId=message.variableHeader().packetId();


        MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(
                MqttMessageType.PUBLISH,
                isDump,
                subQos,isRetain, 0);

        MqttPublishVariableHeader variableHeader = new MqttPublishVariableHeader(topicName,
                packId);


        return  new MqttPublishMessage(mqttFixedHeader, variableHeader, message.content());
    }
}
